#include "zookit.h"
#include "message.h"
#include "rf/event/RFEventManager.h"
#include "rf/event/RFTimerFired.h"

namespace ProcessRouter {

static void zktest_watcher_g(zhandle_t* zh, int type, int state,
        const char* path, void* watcherCtx)
{
    /*
    printf("Something happened.\n");
    printf("type: %d\n", type);
    printf("state: %d\n", state);
    printf("path: %s\n", path);
    printf("watcherCtx: %s\n", (char *)watcherCtx);
    */
}

ZooKit::ZooKit(std::string group)
    : m_handle(NULL)
    , m_group("/spider_v2/"+group)
{
    zoo_set_debug_level(ZOO_LOG_LEVEL_WARN);
}

ZooKit::~ZooKit()
{
    destroy();
    zookeeper_close(m_handle);
}

void ZooKit::connect(std::string host)
{
    m_host = host;
    m_handle = zookeeper_init(host.c_str(), zktest_watcher_g, 30000, 0, (void*)"spider webkit core", 0);

    if (m_handle == NULL) {
        fprintf(stderr, "Error when connecting to zookeeper servers...\n");
        exit(EXIT_FAILURE);
    }
}

void ZooKit::init(std::string nodename)
{
    if (nodename.find("/") != -1)
        fprintf(stderr, "WARRNING protocol contains \"/\", mey case zookeeper path error.\n");

    if (nodename != "") {
        m_path.append(m_group);
        m_path.append("/");
        m_path.append(nodename);
    }

    char new_path[256] = {0};

    std::string i = "happy";
    int ret = zoo_create(m_handle, "/spider_v2", "root", 5,
            &ZOO_OPEN_ACL_UNSAFE, 0, new_path, 256);
    if (ret && ret != ZNODEEXISTS)
        fprintf(stderr, "line: %d Error %d for %s\n", __LINE__, ret, "acreate");

    ret = zoo_create(m_handle, m_group.c_str(), "group", 6,
            &ZOO_OPEN_ACL_UNSAFE, 0, new_path, 256);
    if (ret && ret != ZNODEEXISTS)
        fprintf(stderr, "line: %d Error %d for %s\n", __LINE__, ret, "acreate");

    ret = zoo_create(m_handle, m_path.c_str(), i.c_str(), i.size(),
            &ZOO_OPEN_ACL_UNSAFE, 0, new_path, 256);
    if (ret && ret != ZNODEEXISTS) {
        fprintf(stderr, "line: %d Error %d for %s\n", __LINE__, ret, "acreate");
        exit(EXIT_FAILURE);
    }

    if (ret != ZNODEEXISTS)
        m_path = new_path;

    // 信息节点
    std::string stateinfoPath = m_path + "/stateinfo";
    ret = zoo_create(m_handle, stateinfoPath.c_str(), i.c_str(), i.size(),
            &ZOO_OPEN_ACL_UNSAFE, ZOO_EPHEMERAL, new_path, 256);  // 会话内有效
    if (ret && ret != ZNODEEXISTS) {
        fprintf(stderr, "line: %d Error %d for %s\n", __LINE__, ret, "acreate");
        exit(EXIT_FAILURE);
    }

    std::string pageinfoPath = m_path + "/pageinfo";
    ret = zoo_create(m_handle, pageinfoPath.c_str(), i.c_str(), i.size(),
            &ZOO_OPEN_ACL_UNSAFE, ZOO_EPHEMERAL, new_path, 256);  // 会话内有效
    if (ret && ret != ZNODEEXISTS) {
        fprintf(stderr, "line: %d Error %d for %s\n", __LINE__, ret, "acreate");
        exit(EXIT_FAILURE);
    }
}
 

/*
void ZooKit::watchGroup()
{
}
*/

void ZooKit::destroy()
{
    if (m_path == "")
        return ;

    struct Stat stat;
    int ret = 0;

    // remove info node
    std::string path = m_path + "/stateinfo";
    ret = zoo_exists(m_handle, path.c_str(), 0, &stat);
    if (ret)
        fprintf(stderr, "Error %d for %s\n", ret, "exists");

    ret = zoo_delete(m_handle, path.c_str(), -1);
    if (ret)
        fprintf(stderr, "Error %d for %s\n", ret, "delete");


    path = m_path + "/pageinfo";
    ret = zoo_exists(m_handle, path.c_str(), 0, &stat);
    if (ret)
        fprintf(stderr, "Error %d for %s\n", ret, "exists");

    ret = zoo_delete(m_handle, path.c_str(), -1);
    if (ret)
        fprintf(stderr, "Error %d for %s\n", ret, "delete");

    // remove app node
    ret = zoo_exists(m_handle, m_path.c_str(), 0, &stat);
    if (ret)
        fprintf(stderr, "Error %d for %s\n", ret, "exists");

    ret = 0;
    ret = zoo_delete(m_handle, m_path.c_str(), -1);
    if (ret)
        fprintf(stderr, "Error %d for %s\n", ret, "delete");
}

void ZooKit::set(std::string content)
{
    std::string path = m_path + "/stateinfo";
    int ret = zoo_set(m_handle, path.c_str(), content.data(), content.size(), -1);
    if (ret)
        fprintf(stderr, "Error %d for %s path %s\n", ret, "set", path.c_str());

    if (ret == ZINVALIDSTATE || ret == ZNONODE) {
        destroy();
        zookeeper_close(m_handle);
        connect(m_host);
        init("");
    }
}

void ZooKit::setPageInfo(std::string content)
{
    std::string path = m_path + "/pageinfo";
    int ret = zoo_set(m_handle, path.c_str(), content.data(), content.size(), -1);
    if (ret)
        fprintf(stderr, "Error %d for %s path %s\n", ret, "set", path.c_str());

    if (ret == ZINVALIDSTATE || ret == ZNONODE) {
        destroy();
        zookeeper_close(m_handle);
        connect(m_host);
        init("");
    }
}

bool ZooKit::get(std::string &content, int retry)
{
    char buffer[2048] = {0};
    int len = 0;

    struct Stat stat;

    const int BUF_LEN = 10 * 1024;
    char buf[BUF_LEN] = {0};
    len = BUF_LEN - 1;
    memset(&stat, 0, sizeof(struct Stat));
    int rc = ZOK;
    do {
        len = BUF_LEN - 1;
        rc = zoo_get(m_handle, m_path.c_str(), 0, buf, &len, &stat);
    } while (rc != ZOK && retryFeasible(rc) && retry-- > 0);
    if (rc != ZOK) {
        std::cout << "Failed fetching data: " << zerror(rc) << std::endl;
        return false;
    } else {
        content.append(buf, len);
        return true;
    }
}

bool ZooKit::retryFeasible(int zkrc) {
    if (zkrc == ZCONNECTIONLOSS
            || zkrc == ZMARSHALLINGERROR
            || zkrc == ZOPERATIONTIMEOUT)
        return true;
    else
        return false;
}

}

